package com.zr.infrastructure.utils;

import com.zr.domain.bar.entity.MarketDataEvent;
import com.zr.domain.bar.service.fun.KLineCalculationProcessFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class BarAlgoUtil {

    public static DataStream<MarketDataEvent> calculateKLine(DataStream<MarketDataEvent> inputStream, Time windowSize) {
        return inputStream
                .keyBy(BarAlgoUtil::getBarGroup)
                .window(TumblingEventTimeWindows.of(windowSize))
                .process(new KLineCalculationProcessFunction());
    }


    public static String getBarGroup(MarketDataEvent marketData) {
        return marketData.getChannel() + "_" + marketData.getInstrument() + "_" + marketData.getContract();
    }

    public static String getRedisKey(MarketDataEvent kLine) {
        return kLine.getChannel() + "_" + kLine.getInstrument() + "_"+ kLine.getContract() + "_" + kLine.getTimestamp();
    }


}
